package realtime.app.dws;/*
 *@ClassName : Dws_Test_Paper_TestPaper
 *@Description : $ {description}
 *@Author : NingYu
 *@Date : 2023-03-24 19:{MINUTE}
 */

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import realtime.bean.TestPaperExamBean;
import realtime.func.AsyncDIMFunction;
import realtime.util.ClickHouseUtil;
import realtime.util.DateFormatUtil;
import realtime.util.MyKafkaUtil;

import java.math.BigDecimal;
import java.time.Duration;
import java.util.concurrent.TimeUnit;

public class Dws_Test_Paper_TestPaper {
    public static void main(String[] args) throws Exception {

        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //TODO 2.读取Kafka DWD层 加购主题数据
        String topic = "dwd_test_exam";
        String groupId = "test_paper_0325_04";
        DataStreamSource<String> kafkaDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, groupId));

        //TODO 3.将数据转换为JSON对象
        SingleOutputStreamOperator<JSONObject> jsonDS = kafkaDS.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String s, Collector<JSONObject> collector) throws Exception {
                if (s != null) {
                    try {
                        JSONObject jsonObject = JSONObject.parseObject(s);
                        collector.collect(jsonObject);
                    } catch (Exception e) {
                        System.out.println(s+" : 为空数据,无法转换为json格式");
                    }
                }
            }
        });

        //TODO 4.提取时间生成Watermark
        SingleOutputStreamOperator<JSONObject> jsonWithWMDS = jsonDS
                .assignTimestampsAndWatermarks(WatermarkStrategy.<JSONObject>forBoundedOutOfOrderness(Duration.ofSeconds(2))
                        .withTimestampAssigner(new SerializableTimestampAssigner<JSONObject>() {
                            @Override
                            public long extractTimestamp(JSONObject jsonObject, long l) {
                                return jsonObject.getLong("create_time");
                            }
                        }));

        //TODO 5.按照 user_id进行分组

        KeyedStream<JSONObject, String> keyedByPaperIdDS = jsonWithWMDS.keyBy(json -> json.getString("user_id"));

        //TODO 6.去重并转换为JavaBean对象
        SingleOutputStreamOperator<TestPaperExamBean> beanDS = keyedByPaperIdDS.flatMap(new RichFlatMapFunction<JSONObject, TestPaperExamBean>() {

            private ValueState<String> testIdState;

            @Override
            public void open(Configuration parameters) throws Exception {

                testIdState = getRuntimeContext().getState(new ValueStateDescriptor<String>("testId-state", String.class));
            }

            @Override
            public void flatMap(JSONObject jsonObject, Collector<TestPaperExamBean> collector) throws Exception {

                String lastTestId = testIdState.value();//状态中paperId

                String curTestId = jsonObject.getString("id");// 当前paperId

//                Long personCt = 0L;
//                Double scoreSum = 0.00;
//                Double durationSum = 0.00;

                collector.collect(TestPaperExamBean.builder()
                        .personCt(1l)
                        .paperId(jsonObject.getString("paper_id"))
                        .durationSum(jsonObject.getDouble("duration_sec"))
                        .scoreSum(jsonObject.getDouble("score"))
                        .ts(jsonObject.getLong("submit_time"))
                        .build());
            }
        });

        //TODO 7.开窗、聚合
        SingleOutputStreamOperator<TestPaperExamBean> reduceDS = beanDS
                .keyBy(TestPaperExamBean::getPaperId)
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .reduce(new ReduceFunction<TestPaperExamBean>() {
                    @Override
                    public TestPaperExamBean reduce(TestPaperExamBean v1, TestPaperExamBean v2) throws Exception {
                        v1.setPersonCt(v1.getPersonCt() + v2.getPersonCt());
                        v1.setScoreSum(v1.getScoreSum() + v2.getScoreSum());
                        v1.setDurationSum(v1.getDurationSum() + (v2.getDurationSum()));

                        return v1;
                    }
                }, new WindowFunction<TestPaperExamBean, TestPaperExamBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow timeWindow, Iterable<TestPaperExamBean> iterable, Collector<TestPaperExamBean> collector) throws Exception {
                        TestPaperExamBean next = iterable.iterator().next();

                        next.setScoreAvg(next.getScoreSum()/next.getPersonCt());
                        next.setDurationAvg(next.getDurationSum()/(next.getPersonCt()));

                        next.setStt(DateFormatUtil.toYmdHms(timeWindow.getStart()));
                        next.setEdt(DateFormatUtil.toYmdHms(timeWindow.getEnd()));
                        next.setTs(System.currentTimeMillis());

                        collector.collect(next);
                    }
                });

        //todo 聚合后关联维表
        SingleOutputStreamOperator<TestPaperExamBean> resultDS = AsyncDataStream.unorderedWait(
                reduceDS
                , new AsyncDIMFunction<TestPaperExamBean>("dim_test_paper".toUpperCase()) {
                    @Override
                    public void join(TestPaperExamBean input, JSONObject dimInfo) {
                        input.setPaperName(dimInfo.getString("paper_title".toUpperCase()));
                    }

                    @Override
                    public String getKey(TestPaperExamBean input) {
                        return input.getPaperId();
                    }
                }
                , 60
                , TimeUnit.SECONDS
        );


        //TODO 8.将数据写出
        resultDS.print("Dws_Test_Paper_TestPaper>>>>>>>>\n");

        resultDS.addSink(ClickHouseUtil.getSinkFunction("insert into dws_test_paper_testpaper_window values(?,?,?,?,?,?,?,?,?,?)"));

        //TODO 9.启动任务
        env.execute();
    }
}
